package com.atguigu.edu.realtime.dim.function;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.bean.TableProcessDim;
import com.atguigu.edu.realtime.common.util.JdbcUtil;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.util.*;

public class TableProcessFunction extends BroadcastProcessFunction<JSONObject, TableProcessDim, Tuple2<JSONObject, TableProcessDim>> {

    MapStateDescriptor<String, TableProcessDim> mapDescriptor;

    //map 存状态
    Map<String, TableProcessDim> configMap = new HashMap<>();

    public TableProcessFunction(MapStateDescriptor<String, TableProcessDim> mapDescriptor) {
        this.mapDescriptor = mapDescriptor;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        //将配置表的配置信息预加载到状态中
        Connection mySqlConnection = JdbcUtil.getMysqlConnection();
        List<TableProcessDim> tableProcessDims = JdbcUtil.queryList(mySqlConnection, "select * from edu0422_config.table_process_dim", TableProcessDim.class, true);
        for (TableProcessDim tableProcessDim : tableProcessDims) {
            configMap.put(tableProcessDim.getSourceTable(), tableProcessDim);
        }
        JdbcUtil.closeMysqlConnection(mySqlConnection);
    }

    @Override
    public void processElement(JSONObject jsonObject, BroadcastProcessFunction<JSONObject, TableProcessDim, Tuple2<JSONObject, TableProcessDim>>.ReadOnlyContext readOnlyContext, Collector<Tuple2<JSONObject, TableProcessDim>> collector) throws Exception {
        //获取当前数据库的表名
        String table = jsonObject.getString("table");
        //获取广播状态
        ReadOnlyBroadcastState<String, TableProcessDim> broadcastState = readOnlyContext.getBroadcastState(mapDescriptor);
        //根据表名到广播状态中获取配置信息
        TableProcessDim tableProcessDim = null;
        //判断是否是维度数据
        if ((tableProcessDim = broadcastState.get(table)) != null
                || (tableProcessDim = configMap.get(table)) != null) {
            JSONObject dataJSONObj = jsonObject.getJSONObject("data");
            //过滤不需要传递的数据
            String sinkColumns = tableProcessDim.getSinkColumns();
            deleteNoNeedColumn(dataJSONObj, sinkColumns);

            //补充操作类型
            String type = jsonObject.getString("type");
            dataJSONObj.put("type", type);

            collector.collect(Tuple2.of(dataJSONObj, tableProcessDim));
        }
    }

    @Override
    public void processBroadcastElement(TableProcessDim tableProcessDim, BroadcastProcessFunction<JSONObject, TableProcessDim, Tuple2<JSONObject, TableProcessDim>>.Context context, Collector<Tuple2<JSONObject, TableProcessDim>> collector) throws Exception {
        String op = tableProcessDim.getOp();

        BroadcastState<String, TableProcessDim> broadcastState = context.getBroadcastState(mapDescriptor);
        String sourceTable = tableProcessDim.getSourceTable();
        if ("d".equals(op)) {
            broadcastState.remove(sourceTable);
            configMap.remove(sourceTable);
        } else {
            broadcastState.put(sourceTable, tableProcessDim);
            configMap.put(sourceTable, tableProcessDim);
        }
    }

    //过滤不需要的字段
    private static void deleteNoNeedColumn(JSONObject dataJSONObject, String sinkColumns) {
        List<String> list = Arrays.asList(sinkColumns.split(","));
        Set<Map.Entry<String, Object>> entries = dataJSONObject.entrySet();
        entries.removeIf(entry -> !list.contains(entry.getKey()));
//        Iterator<Map.Entry<String, Object>> iterator = entries.iterator();
//        while (iterator.hasNext()){
//            if(!list.contains(iterator.next())){
//                iterator.remove();
//            }
//        }
    }
}